package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.StockBean;
import cn.itcast.bean.StockIncrBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;

/**
 * @Date 2021
 * 个股涨跌幅窗口函数处理
 */
//1.新建StockUpDownWindowFunction 窗口函数
public class StockUpDownWindowFunction extends RichWindowFunction<CleanBean, StockIncrBean, String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow window, Iterable<CleanBean> input, Collector<StockIncrBean> out) throws Exception {
        /**
         * 2.记录最新个股
         * 3.格式化日期
         * 4.指标计算
         *   涨跌、涨跌幅、振幅
         * 5.封装输出数据
         */
        //2.记录最新个股
        CleanBean line = null;
        for (CleanBean cleanBean : input) {
            if (line == null) {
                line = cleanBean;
            }
            if (cleanBean.getEventTime() > line.getEventTime()) {
                line = cleanBean;
            }
            //3.格式化日期
            Long transfer = DateUtil.longTimeTransfer(line.getEventTime(), Constant.format_YYYYMMDDHHMMSS);
            //* 4.指标计算
            //         *   涨跌、涨跌幅、振幅
            /**
             * 今日涨跌=当前价-前收盘价
             * 今日涨跌幅（%）=（当前价-前收盘价）/ 前收盘价 * 100%
             * 今日振幅 =（当日最高点的价格－当日最低点的价格）/昨天收盘价×100%
             */
            //对金额的计算,要考虑精度,一般都是采用BigDecimal类型
            //今日涨跌=当前价-前收盘价
            BigDecimal subtract = line.getTradePrice().subtract(line.getPreClosePrice());
            //保留两位小数,四舍五入
            //日涨跌幅（%）=（当前价-前收盘价）/ 前收盘价 * 100%
            BigDecimal increase = subtract.divide(line.getPreClosePrice(),2,BigDecimal.ROUND_HALF_UP);
            //今日振幅 =（当日最高点的价格－当日最低点的价格）/昨天收盘价×100%
            BigDecimal divide = (line.getMaxPrice().subtract(line.getMinPrice())).divide(line.getPreClosePrice(), 2, BigDecimal.ROUND_HALF_UP);
            //5.封装输出数据
            //eventTime、secCode、secName、increase、tradePrice、updown、tradeVol、amplitude、
            //preClosePrice、tradeAmt、tradeTime、source
            StockIncrBean stockIncrBean = new StockIncrBean();
            stockIncrBean.setEventTime(line.getEventTime());
            stockIncrBean.setSecCode(line.getSecCode());
            stockIncrBean.setSecName(line.getSecName());
            stockIncrBean.setIncrease(increase);
            stockIncrBean.setTradePrice(line.getTradePrice());
            stockIncrBean.setUpdown(subtract);
            stockIncrBean.setTradeVol(line.getTradeVolumn());
            stockIncrBean.setAmplitude(increase);
            stockIncrBean.setPreClosePrice(line.getPreClosePrice());
            stockIncrBean.setTradeAmt(line.getTradeAmt());
            stockIncrBean.setTradeTime(transfer);
            stockIncrBean.setSource(line.getSource());
            out.collect(stockIncrBean);
        }
    }
}
